/**
 * 
 */
package edu.cornell.cs.lsi.mapreduce.pass3;

import java.io.IOException;
import java.util.ArrayList;
import java.util.Collections;
import java.util.Comparator;
import java.util.Hashtable;
import java.util.List;
import java.util.SortedSet;
import java.util.StringTokenizer;
import java.util.TreeSet;
import java.util.Map.Entry;

import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.Job;
import org.apache.hadoop.mapreduce.Mapper;
import org.apache.hadoop.mapreduce.Reducer;
import org.apache.hadoop.mapreduce.lib.input.FileInputFormat;
import org.apache.hadoop.mapreduce.lib.output.FileOutputFormat;

import edu.cornell.cs.lsi.mapreduce.pass1.DoublePoint;
import edu.cornell.cs.lsi.mapreduce.pass2.Point;
import edu.cornell.cs.lsi.mapreduce.pass2.UnionFindImpl;

/**
 * @author poornima
 *
 */
public class MapReduceThirdPass {
	
	private static final double BORDERLESS_COL_WIDTH = .1f;
	private static final double Lmax = .09f;
	
	private static final double Lmax_SQUARED = Lmax * Lmax; 
	private static final double G = Lmax + BORDERLESS_COL_WIDTH;
	
	private static final double MIN_WEIGHT = 0.0f;
	private static final double MAX_WEIGHT = 1.0f;
	private static final double SIZE_OF_GRID = 1.0f; 
	
	private static final String THIRD_PASS = "third pass";
	
	private static Hashtable<Point, Point> uniqueEntries = new Hashtable<Point, Point>();
	

	
	/**
	 * Union find implementation
	 */
//	private static UnionFindImpl ufImpl = new UnionFindImpl();	

	/**
	 * Extending the <code>MapReduceBase</code> class and implementing the <code>Mapper</code>.
	 * @author poornima
	 */
    public static class Map extends Mapper<Object, Text, Text, Text> {
    	
    /**
     * Overriding the map() method. The mapper passes the input points as it is to the reducer.
     * input to the mapper in the form: <x1><space><y1><space><x2><space><y2><space>
     */
     @Override	
     public void map(Object key, Text value, Context context
     ) throws IOException, InterruptedException {
    	 // read the line and tokenize
        String line = value.toString();
        StringTokenizer tokenizer = new StringTokenizer(line);
        List<String> tokens = new ArrayList<String>();
        while (tokenizer.hasMoreTokens()) {
        	tokens.add(tokenizer.nextToken());        
        }
        if(tokens != null && tokens.size() == 3){
        	// this  if from the production file with the vertices.
        	double x = Double.parseDouble(tokens.get(0));
			double y = Double.parseDouble(tokens.get(1));
			double weight = Double.parseDouble(tokens.get(0));
			int colNum;
			double offset;
			String pointStr;
			if(MIN_WEIGHT <= weight && MAX_WEIGHT > weight){
				colNum = (int)(x / G);
			//	System.out.println("---------- colNum :"+colNum+" total column count "+getColCount());
				offset = x - ((double)colNum) * G;
				pointStr = x + " " + y;
				if(offset > G - Lmax){
				//	System.out.println("---------- colNum :"+(colNum+1)+" pointStr "+pointStr);
					context.write(new Text(String.valueOf(colNum + 1)), new Text(pointStr));
				}
				//System.out.println("---------- colNum :"+(colNum)+" pointStr "+pointStr);
				context.write(new Text(String.valueOf(colNum)), new Text(pointStr));
			}        	
        }
        Text outputKey;
        Text outputValue;
        
        if(tokens != null && tokens.size() == 4){
        	int columnCnt = getColCount();
        	for(int count = 0 ; count < columnCnt;count++){
        		outputKey = new Text(String.valueOf(count));
        		outputValue = new Text(tokens.get(0)+" "+tokens.get(1)+" "+tokens.get(2)+" "+tokens.get(3));
        	//	System.out.println("---------- colNum :"+(outputKey)+" vertex:root "+outputValue);
        		context.write(outputKey, outputValue);
        	}        	
        }        
      }      
    }
    
    private static int getColCount(){
    	double result = SIZE_OF_GRID / G;
    	if((double)((int)result) == result) return ((int)result);
    	return ((int)result) + 1;
    }
    
    /**
     * Extending the <code>MapReduceBase</code> class and implementing the <code>Reducer</code>.
     * @author poornima
     *
     */
    public static class Reduce extends Reducer<Text, Text, Text, Text> {
    	
      /**
       * Override the Reducer method.
       */
      @Override
      public void reduce(Text key, Iterable<Text> values, 
              Context context
              ) throws IOException, InterruptedException {
    	  System.out.println("---------- DEBUG ------------ : starts");
    	  System.out.println("-- key ------------"+key.toString());
    	  List<DoublePoint> allVertices = new ArrayList<DoublePoint>();
    	  UnionFindImpl ufImpl = new UnionFindImpl();
    	  // read the key and tokenize to extract the key vertex
    	  for(Text value : values){
    		  System.out.println("-------- value: "+value);
    		  StringTokenizer tokenizer = new StringTokenizer(value.toString());
    		  List<String> tokens = new ArrayList<String>();
    		  while(tokenizer.hasMoreTokens()){
    			  tokens.add(tokenizer.nextToken());    			  
    		  }
    		  // populate the vertices in theh set
    		  if(tokens != null && tokens.size() == 2){
    			  // this is a vertex
    			  System.out.println("-------- vertex : ----------- ("+tokens.get(0)+","+tokens.get(1)+")");
    			  allVertices.add(new DoublePoint(Double.parseDouble(tokens.get(0)),Double.parseDouble(tokens.get(1))));    			  
    		  }
    		  // populate the edges from the second pass in the UF table
    		  if(tokens != null && tokens.size() == 4){
    			  System.out.println("-------- from pass 2 output ------------");
    			  System.out.println("-------- key vertex : ----------- ("+tokens.get(0)+","+tokens.get(1)+")");
    			  System.out.println("-------- root vertex : ----------- ("+tokens.get(2)+","+tokens.get(3)+")");
    			  Point vertex = new Point(Double.parseDouble(tokens.get(0)),Double.parseDouble(tokens.get(1)));
    			  Point root = new Point(Double.parseDouble(tokens.get(2)),Double.parseDouble(tokens.get(3)));
    			  ufImpl.doUnionFind(vertex, root);
    		  }
    	  }
    	  
    	  System.out.println("---- total vertices : from prod file : "+allVertices.size());
    	  System.out.println("---- size of vertices with roots from the second pass : "+ufImpl.getComponentTable().size());
    	  
    	  int colNum = Integer.parseInt(key.toString());
    	  int remIndex;
    	  int seekIndex;
    	  DoublePoint curPoint;
    	  DoublePoint minPoint;
    	  DoublePoint maxPoint;
    	  SortedSet<DoublePoint> pointsInSpan;
    	  double leftColBoundary = ((double)(colNum)) * G;
		  double rightColBoundary = leftColBoundary + BORDERLESS_COL_WIDTH;
		  
    	  // sort the vertices
			Comparator<DoublePoint> yComp = new Comparator<DoublePoint>(){

				@Override
				public int compare(DoublePoint arg0, DoublePoint arg1) {
					if(arg0.getY() < arg1.getY()) return -1;
					if(arg0.getY() > arg1.getY()) return 1;
					if(arg0.getX() < arg1.getX()) return -1;
					if(arg0.getX() > arg1.getX()) return 1;				
					return 0;
				}
				
			};			
			
			Comparator<DoublePoint> xComp = new Comparator<DoublePoint>(){

				@Override
				public int compare(DoublePoint arg0, DoublePoint arg1) {
					if(arg0.getX() < arg1.getX()) return -1;
					if(arg0.getX() > arg1.getX()) return 1;
					if(arg0.getY() < arg1.getY()) return -1;
					if(arg0.getY() > arg1.getY()) return 1;				
					return 0;
				}
				
			};
			
			TreeSet<DoublePoint> tree = new TreeSet<DoublePoint>(yComp);			
			Collections.sort(allVertices, xComp);
			
			remIndex = 0;
			boolean hasNoEdge = true;
			List<Point> singleVertices = new ArrayList<Point>();
			for(seekIndex = 0; seekIndex < allVertices.size(); seekIndex++){
				hasNoEdge = true;
				curPoint = allVertices.get(seekIndex);
				
				minPoint = new DoublePoint(-1f, curPoint.getY() - Lmax);
				maxPoint = new DoublePoint(2f, curPoint.getY() + Lmax);
				pointsInSpan = tree.subSet(minPoint, maxPoint);
				
				System.out.println("--- seek loop: curPoint: "+curPoint.getX()+" "+curPoint.getY());
				System.out.println("--- seek loop: minPoint: "+minPoint.getX()+" "+minPoint.getY());
				System.out.println("--- seek loop: maxPoint: "+maxPoint.getX()+" "+maxPoint.getY());

				
				for(DoublePoint pointInSpan : pointsInSpan){
					
					System.out.println("--- seek loop: pointInSpan: "+pointInSpan.getX()+" "+pointInSpan.getY());
					if(pointInSpan.withinRange(curPoint, Lmax_SQUARED)){
						System.out.println(" within the range ");
						// if we're keeping track of # of edges, this is where we would do something
						
						if(pointInSpan.getX() > leftColBoundary || curPoint.getX() > leftColBoundary){
							Point pt1 = new Point(pointInSpan.getX(),pointInSpan.getY());
							Point pt2 = new Point(curPoint.getX(),curPoint.getY());
							// do union find for the discovered edge.
							ufImpl.doUnionFind(pt1, pt2);
							System.out.println("---- edge discovered: pt1:("+pt1.getX()+","+pt1.getY()+")"+"  pt2:("+pt2.getX()+","+pt2.getY()+")");
							hasNoEdge = false;
						}						
						pointInSpan.union(curPoint);
					}
				}// end of for loop
				
				if(seekIndex + 1 < allVertices.size()){
					while(remIndex < seekIndex && allVertices.get(remIndex).getX() < curPoint.getX() - Lmax){
						tree.remove(allVertices.get(remIndex));
						remIndex++;
					}
				}
				tree.add(curPoint);
				if(hasNoEdge){
					System.out.println("---- single vertex: "+curPoint.getX()+" "+curPoint.getY());
					singleVertices.add(new Point(curPoint.getX(),curPoint.getY()));
				}
			}// end of for loop		
			
			// check if the single vertices that are left out are connected to either of the connected components from Pass 2
			Hashtable<Point, Point> outputPairs = new Hashtable<Point, Point>();
			for(Entry<Point,Point> entry : ufImpl.getComponentTable().entrySet()){
				Point vertex = entry.getKey();
				Point root = entry.getValue();
				for(Point singleVertex : singleVertices){
					if(withinRange(vertex,singleVertex,Lmax_SQUARED)){
						ufImpl.doUnionFind(singleVertex, vertex);
					}else if(withinRange(root,singleVertex,Lmax_SQUARED)){
						ufImpl.doUnionFind(root, vertex);
					}
				}
				//outputPairs.put(vertex, root);
			}		
			
			
			// output the vertex and the root vertex..
			Text outputKey;
			Text outputValue;				
			for(Entry<Point,Point> entry : ufImpl.getComponentTable().entrySet()){
				Point vertex = entry.getKey();
				Point root = entry.getValue();
				//if((vertex.getX() > leftColBoundary && vertex.getX() < rightColBoundary) || (vertex.getX() == rightColBoundary)){
				if((vertex.getX() > leftColBoundary)){
					if(uniqueEntries.contains(vertex)){
						continue;
					}
					uniqueEntries.put(vertex, root);
					outputKey = new Text(vertex.getX()+" "+vertex.getY());
					outputValue = new Text(root.getX()+" "+root.getY());
					context.write(outputKey, outputValue);
					
				}				
			}
			System.out.println("--------------- display map ------------------");
			ufImpl.displayMap();
			System.out.println("--------------- display map ------------------");
			System.out.println(" ------------- DEBUG: third pass end ----");
        }     
    }
    
    private static boolean withinRange(Point one, Point two, double rangeSquared){
    	System.out.println("--- single vertex: "+two.getX()+" "+two.getY()+"---- root / vertex: "+one.getX()+" "+one.getY()+"   -- output"+((one.getX() - two.getX()) * (one.getX() - two.getX()) + (one.getY() - two.getY()) * (one.getY() - two.getY()) <= rangeSquared));
    	
		return (one.getX() - two.getX()) * (one.getX() - two.getX()) + (one.getY() - two.getY()) * (one.getY() - two.getY()) <= rangeSquared;
	}
    
    

    public static void main(String[] args) throws Exception {
    	if(args == null || args.length != 2){
    		System.out.println("Please pass the input and the output folder names.");
    		return;
    	}
    	Configuration conf = new Configuration();
        Job job = new Job(conf, THIRD_PASS);
        job.setJarByClass(MapReduceThirdPass.class);
        job.setMapperClass(Map.class);
        //job.setCombinerClass(IntSumReducer.class);
        job.setReducerClass(Reduce.class);
        job.setOutputKeyClass(Text.class);
        job.setOutputValueClass(Text.class);
        FileInputFormat.addInputPath(job, new Path(args[0]));
        FileOutputFormat.setOutputPath(job, new Path(args[1]));
        System.exit(job.waitForCompletion(true) ? 0 : 1);
    }
}
